﻿/*
** Xin YUAN, 2023, BSD (2)
*/

////////////////////////////////////////////////////////////////////////////////

#include "base/_x_thread_pool.h"

#include "sys_sync.h"

////////////////////////////////////////////////////////////////////////////////

//work_data_base

class work_data_base
{
protected:
	work_data_base() noexcept
	{
	}
	work_data_base(const work_data_base&) = delete;
	~work_data_base() noexcept
	{
	}
	work_data_base& operator=(const work_data_base&) = delete;

private:
	friend class task_base;
};

//task_base

class task_base
{
private:
	struct work_item
	{
		void* p;
	};

	enum {
		CORE_NUM = 4,
		FREE_NUM = 2
	};

protected:
	task_base() noexcept : m_iCallFlag(0), m_iExit(0), m_iFinish(0), m_iShutdown(0),
					m_uPoolNum(0), m_uSleepNum(0)
	{
	}
	task_base(const task_base&) = delete;
	~task_base() noexcept
	{
		assert( !is_valid() );
	}

	task_base& operator=(const task_base&) = delete;

protected:
	bool init(task_util::task_callback cb) noexcept
	{
		assert( !is_valid() );
		//callback
		m_util.callback = cb;
		//work queue
		m_queue.Clear();
		if ( m_arrWork.IsEmpty() ) {
			if ( !m_arrWork.SetSize(32) )
				return false;
			m_queue.SetCapacity(m_arrWork.GetSize());
		}
		//mutex & cond
		if ( !m_mutex.IsValid() ) {
			if ( !m_mutex.Init() )
				return false;
			m_cond.Init();
		}
		//thread pool
		m_uSleepNum = 0;
		if ( !init_all_threads() )
			return false;
		return true;
	}

public:
	void Shutdown() noexcept
	{
		if ( is_valid() ) {
			//thread pool shutdown
			dump_all_threads();
		}
	}

	bool SubmitWork(work_data_base* pw) noexcept
	{
		assert( is_valid() );
		if ( __sync_val_compare_and_swap(&m_iExit, 1, 1) )
			return true;
		m_mutex.Lock();
		if ( !post_work(pw) ) {
			m_mutex.Unlock();
			return false;
		}
		m_mutex.Unlock();
		m_cond.Signal();  //may have no effect
		return true;
	}

private:
	static void* worker_thread(void* p) noexcept
	{
		task_base* pThis = (task_base*)p;
		__sync_val_compare_and_swap(&pThis->m_iFinish, 0, 1);
		pThis->process_work();
		return NULL;
	}

	void process_work() noexcept
	{
		m_mutex.Lock();
		while ( m_iShutdown == 0 ) {
			//task
			uintptr_t wnum = m_queue.GetSize();
			if ( wnum <= m_uPoolNum - FREE_NUM )
				++ m_uSleepNum;
			else
				m_uSleepNum = 0;
			if ( m_uSleepNum > m_uPoolNum && m_uPoolNum > CORE_NUM ) {
				//check thread-decreasing
				m_mutex.Unlock();
				if ( !__sync_val_compare_and_swap(&m_iExit, 1, 1) )
					dump_worker_thread();  //no check
				m_mutex.Lock();
			}
			else if ( m_uPoolNum < _POOL_MAX_THREAD_NUM && m_uPoolNum < (uint32_t)(wnum / 2) ) {
				//check thread-increasing
				m_mutex.Unlock();
				if ( !__sync_val_compare_and_swap(&m_iExit, 1, 1) )
					init_thread(&worker_thread, this);  //no check
				m_mutex.Lock();
			}
			//empty
			if ( m_queue.IsEmpty() ) {
				m_cond.Wait(m_mutex);  //block
				continue;
			}
			//work
			work_data_base* pw = (work_data_base*)(m_arrWork.GetAt(m_queue.De()).p);
			m_mutex.Unlock();
			if ( !__sync_val_compare_and_swap(&m_iExit, 1, 1) )
				m_util.callback(pw, this);
			m_mutex.Lock();
		}
		//exit
		m_last_ptd = ::pthread_self();
		m_iShutdown = 0;
		m_mutex.Unlock();
	}

//tools
	bool post_work(work_data_base* pw) noexcept
	{
		if ( m_queue.IsFull() ) {
			if ( !m_arrWork.InsertAt(m_queue.GetRear()) )
				return false;
			m_queue.Add();
		}
		work_item& item = m_arrWork.GetAt(m_queue.En());
		item.p = pw;
		return true;
	}

	//-1 : error  0 : succeeded  1 : discard
	int init_thread(void*(*func)(void*), void* pv) noexcept
	{
		//acquire
		if ( __sync_val_compare_and_swap(&m_iCallFlag, 0, 1) )
			return 1;
		//reset
		__sync_val_compare_and_swap(&m_iFinish, 1, 0);
		//create
		pthread_t ptd;
		int res = ::pthread_create(&ptd, NULL, func, pv);
		if ( res != 0 ) {
			//release
			__sync_val_compare_and_swap(&m_iCallFlag, 1, 0);
			return -1;
		}
		//wait
		while ( __sync_val_compare_and_swap(&m_iFinish, 1, 1) != 1 ) {
			thread_sleep(1);  //1ms
		}
		//adjust
		m_mutex.Lock();
		++ m_uPoolNum;
		m_mutex.Unlock();
		//release
		__sync_val_compare_and_swap(&m_iCallFlag, 1, 0);
		return 0;
	}
	bool init_all_threads() noexcept
	{
		m_uPoolNum = 0;
		for ( int i = CORE_NUM; i > 0; -- i ) {
			int ret = init_thread(&worker_thread, this);
			assert( ret != 1 );
			if ( ret != 0 ) {
				dump_all_threads();
				return false;
			}
		}
		return true;
	}

	// 0 : succeeded   1 : discard
	int dump_worker_thread() noexcept
	{
		//acquire
		if ( __sync_val_compare_and_swap(&m_iCallFlag, 0, 1) )
			return 1;
		//set
		m_mutex.Lock();
		m_iShutdown = 1;
		m_cond.Signal();
		//wait
		while ( m_iShutdown != 0 ) {
			m_mutex.Unlock();
			thread_sleep(1);  //1ms
			m_mutex.Lock();
		}
		m_mutex.Unlock();
		//join
		void* result;
		int res = ::pthread_join(m_last_ptd, &result);
		(void)res;
		assert( res == 0 );
		//adjust
		m_mutex.Lock();
		-- m_uPoolNum;
		m_uSleepNum = 0;
		m_mutex.Unlock();
		//release
		__sync_val_compare_and_swap(&m_iCallFlag, 1, 0);
		return 0;
	}
	void dump_all_threads() noexcept
	{
		__sync_val_compare_and_swap(&m_iExit, 0, 1);
		m_mutex.Lock();
		while ( m_uPoolNum != 0 ) {
			m_mutex.Unlock();
			dump_worker_thread();  //no check
			thread_sleep(1);  //1ms
			m_mutex.Lock();
		}
		m_mutex.Unlock();
		__sync_val_compare_and_swap(&m_iExit, 1, 0);
	}

private:
	bool is_valid() const noexcept
	{
		return m_uPoolNum != 0;
	}

private:
	sync_mutex m_mutex;
	sync_condition m_cond;

	pthread_t m_last_ptd;

	volatile int m_iCallFlag;

	volatile int m_iExit;
	volatile int m_iFinish;

	//conditions
	volatile int m_iShutdown;

	volatile uint32_t m_uPoolNum;
	volatile uint32_t m_uSleepNum;

	CTL::PlainArray<work_item> m_arrWork;
	CTL::CyclicQueueUtility<uintptr_t> m_queue;

	//utility
	task_util m_util;
};

////////////////////////////////////////////////////////////////////////////////
